package com.gitee.filter.bean

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.gitee.etl.process.bean.MQETL
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}

abstract class BeanFilter(envs: StreamExecutionEnvironment) extends MQETL(envs) {
  /**
    * 处理数据的接口,不要这个方法
    */
  override def process(): Unit = {}

  /**
    * 爬虫识别程序的接口
    */
  def recognize(): Unit

  /**
    * 发送爬虫ip给mysql
    */
  def sendMysql(): RichSinkFunction[(String, Long)] = {
    new RichSinkFunction[(String, Long)] {
      var conn: Connection = _
      var ps: PreparedStatement = _

      //执行保存数据逻辑
      override def invoke(value: (String, Long), context: SinkFunction.Context[_]): Unit = {
        ps.setString(1,value._1)
        ps.setLong(2,value._2)
        ps.executeUpdate()
      }

      override def close(): Unit = {
        if (conn != null) conn.close()
        if (ps != null) ps.close()
      }

      override def open(parameters: Configuration): Unit = {
        val sql = "insert into account(name, money) values(?,?)"
        val url = "jdbc:mysql://localhost:3306/day21_db"
        val user = "root"
        val password = "root"
        conn = DriverManager.getConnection(url, user, password)
        ps = conn.prepareStatement(sql)
      }
    }
  }
}
